package org.funtester.performance.books.chapter03.section4;

import org.funtester.performance.books.chapter03.common.DataHandleTool;
import org.funtester.performance.books.chapter03.common.ThreadTool;
import org.funtester.performance.books.chapter03.section3.ThreadTask;

import java.util.List;
import java.util.Vector;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

/**
 *  多线程任务执行器
 */
public class TaskExecutor {

    /**
     * 线程池
     */
    ThreadPoolExecutor poolExecutor;

    /**
     * 总执行次数
     */
    public AtomicInteger executeNumStatistic;

    /**
     * 执行错误次数
     */
    public AtomicInteger errorNumStatistic;

    /**
     * 用于记录所有请求时间
     */
    public Vector<Integer> costTimeStatistic;

    /**
     * 任务描述
     */
    public String taskDesc;

    /**
     * 开始时间,单位毫秒
     */
    public long startTimestamp;

    /**
     * 结束时间,单位毫秒
     */
    public long endTimestamp;

    /**
     * 多线程任务集合
     */
    public List<ThreadTask> tasks;

    /**
     * 用于停止任务的计数器
     */
    public CountDownLatch stopCountDownLatch;

    /**
     * 用于Rump-Up功能,标识线程启动情况
     */
    public CountDownLatch rumpUpCountDownLatch;

    /**
     * Rump-Up的总时间,单位秒
     */
    public int rumpUpTime;

    /**
     * 用于实时统计一段时间总耗时
     */
    public static LongAdder realTimeCostTime = new LongAdder();

    /**
     * 用于实时统计一段时间总次数
     */
    public static LongAdder realTimeCostTimes = new LongAdder();


    /**
     * 用于控制实时统计线程结束开关
     */
    public boolean realTimeKey = true;

    /**
     * 实时信息输出线程,用于实时统计一段时间的TPS和平均耗时
     */
    public Thread realTimeThread;

    /**
     * 无参构造方法
     */
    public TaskExecutor() {
    }

    /**
     * 构造方法
     * @param taskDesc 任务描述
     * @param tasks 多线程任务集合
     * @param rumpUpTime Rump-Up的总时间,单位秒
     */
    public TaskExecutor(List<ThreadTask> tasks, String taskDesc, int rumpUpTime) {
        this.tasks = tasks;// 初始化多线程任务集合
        this.taskDesc = taskDesc;// 初始化任务描述
        this.rumpUpTime = rumpUpTime;// 初始化Rump-Up时间
        this.executeNumStatistic = new AtomicInteger(0);// 初始化执行次数
        this.errorNumStatistic = new AtomicInteger(0);// 初始化执行错误次数
        this.costTimeStatistic = new Vector<>();// 初始化请求时间集合
        this.poolExecutor = ThreadTool.createPool(tasks.size()); // 初始化线程池
        this.stopCountDownLatch = new CountDownLatch(tasks.size());// 初始化停止任务计数器
        this.rumpUpCountDownLatch = new CountDownLatch(tasks.size());// 初始化Rump-Up计数器
        for (int i = 0; i < tasks.size(); i++) {// 初始化多线程任务的CountDownLatch同步类属性
            tasks.get(i).stopCountDownLatch = stopCountDownLatch;
            tasks.get(i).rumpUpCountDownLatch = rumpUpCountDownLatch;
        }
    }

    /**
     * 开始执行任务
     */
    public void start() throws InterruptedException {
        if (realTimeThread == null) realTimeThread = new Thread() {// 创建线程,用于实时统计一段时间的TPS和平均耗时
            @Override
            public void run() {
                while (realTimeKey) {
                    ThreadTool.sleep(1000);// 休眠1秒
                    long sumCost = realTimeCostTime.sumThenReset();// 重置总耗时
                    long sumTimes = realTimeCostTimes.sumThenReset();// 重置总次数
                    System.out.println(String.format("实时统计TPS: %d, 平均耗时: %d", sumTimes, sumTimes == 0 ? 0 : sumCost / sumTimes));
                    int successNum = 0;// 成功次数
                    int errorNum = 0;// 错误次数
                    for (ThreadTask task : tasks) {// 遍历多线程任务集合,累加成功次数和错误次数
                        successNum += task.executeNum;
                        errorNum += task.errorNum;
                    }
                    if (errorNum > successNum) ThreadTask.ABORT.set(true);// 如果错误次数大于成功次数,则终止测试任务
                }
            }
        };
        realTimeThread.start();
        int gap = rumpUpTime * 1000 / tasks.size();// 计算每个线程的启动间隔
        for (ThreadTask task : tasks) {// 遍历多线程任务集合,提交线程池执行
            poolExecutor.execute(task);// 提交线程池执行
            ThreadTool.sleep(gap);// 休眠,间隔提交多线程任务
        }
        rumpUpCountDownLatch.await();// 等待Rump-Up计数器为0
        tasks.forEach(f -> f.countState = true);
        System.out.println("Rump-Up结束,开始执行测试任务!");// 打印Rump-Up结束日志
        this.startTimestamp = System.currentTimeMillis();// 记录开始时间
        stopCountDownLatch.await();// 等待停止任务计数器为0
        this.endTimestamp = System.currentTimeMillis();// 记录结束时间
        this.poolExecutor.shutdown();// 关闭线程池
        realTimeKey = false;// 关闭实时统计线程
        realTimeThread.join();// 等待实时统计线程结束
        handleData();// 处理数据
    }

    /**
     * 处理数据,在测试结束后运行
     */
    public void handleData() {
        for (int i = 0; i < tasks.size(); i++) {
            ThreadTask threadTask = tasks.get(i);
            this.executeNumStatistic.addAndGet(threadTask.executeNum);// 累加执行次数
            this.errorNumStatistic.addAndGet(threadTask.errorNum);// 累加执行错误次数
            this.costTimeStatistic.addAll(threadTask.costTime);// 添加请求时间集合
        }
        int sum = tasks.stream().mapToInt(f -> f.totalNum).sum();
        long costTime = (endTimestamp - startTimestamp) / 1000;// 计算压测时长
        DataHandleTool.Quantile quantile = DataHandleTool.statisticData(costTimeStatistic);
        System.out.println(String.format("测试TPS: %d, 平均耗时: %d", this.tasks.size() * 1000 / quantile.average, quantile.average));// 打印第一种方法统计的测试TPS
        System.out.println(String.format("测试TPS: %d, 总执行次数: %d", executeNumStatistic.get() / costTime, executeNumStatistic.get()));// 打印第二种方法统计的测试TPS
        quantile.print();// 打印请求时间统计结果
        System.out.println(String.format("任务执行完毕! 压测时长: %d 秒, 预期执行次数: %d, 实际执行次数 %d, 错误次数 %d, 耗时收集数量: %d", costTime, sum, executeNumStatistic.get(), errorNumStatistic.get(), costTimeStatistic.size()));// 打印任务执行完毕
    }

    /**
     * 停止执行任务
     */
    public void stop() {
        for (int i = 0; i < tasks.size(); i++) {
            tasks.get(i).needStop = true;
        }
    }


}
